/*
 * Based on:
 *
 * balance - a balancing tcp proxy
 * $Revision: 3.42 $
 *
 * Copyright (c) 2000-2007,2008 by Thomas Obermair (obermair@acm.org)
 * and Inlab Software GmbH (info@inlab.de), Gruenwald, Germany.
 * All rights reserved.
 */

#include "balance.h"

COMMON* common;

static int hashfailover = 0;
static int debugflag = 0;
static int foreground = 0;
static int packetdump = 0;
static int sockbufsize = SOCKET_BUFFER_SIZE;
static char *balance_socket_path = NULL;
static struct timeval sel_tmout = { 0, 0 }; /* seconds, microseconds */
static struct timeval save_tmout = { 0, 0 }; /* seconds, microseconds */

static char rendezvousfile[FILE_NAME_LEN] = "./radezvous.shm";
static int rendezvousfd;
static int shmfilefd;
char workers_path[FILE_NAME_LEN];

int max_retries = 100;
enum boolean_t is_one_worker = FALSE;

unsigned long client_connects = 0;

void printStatistics() {
	fprintf(stderr, "\nnumber of clients %d \n", client_connects);
	fprintf(stderr, "number of failures %d \n", cmn_fc(common));
	int i;
	fprintf(stderr, "number of groups %d \n", cmn_ngroups(common));
	for (i = 0; i < cmn_ngroups(common); i++) {
		int j;
		fprintf(stderr, "\tnumber of channels %d \n", grp_nchannels(common,i));
		for (j = 0; j < grp_nchannels(common,i); j++) {
			fprintf(stderr,
					"\t\tgroup:%d channel:%d tc: %d, bsent: %d bresv: %d \n",
					i, j, chn_tc(common, i,j), chn_bsent(common, i,j),
					chn_breceived(common, i,j));
		}
	}
	fflush(stderr);
}
/*
 * to print statistics
 * ps -x | grep balance
 * kill -s sigusr1 'pid'
 */
void sig_handler(int sig) {
	signal(sig, sig_handler); /* reset it to this function */
	printStatistics();
}

static int fatal_err_dump(char *text) {
	printStatistics();
	perror(text);
	exit(-1);
}

/* locking ... */

int a_readlock(off_t start, off_t len) {
	int rc;
	struct flock fdata;
	fdata.l_type = F_RDLCK;
	fdata.l_whence = SEEK_SET;
	fdata.l_start = 0;
	fdata.l_len = 0;
	// fdata.l_sysid=0;
	// fdata.l_pid=0;
	repeat: if ((rc = fcntl(rendezvousfd, F_SETLKW, &fdata)) < 0) {
		if (errno == EINTR) {
			goto repeat;
			// 8-)
		} else {
			perror("readlock");
			exit(EX_OSERR);
		}
	}
	return (rc);
}

void b_readlock(void) {
	a_readlock(0, 0);
}

void c_readlock(int group, int channel) {
	a_readlock(((char *) &(grp_channel(common, group, channel)))
			- (char *) common, sizeof(CHANNEL));
}

int a_writelock(off_t start, off_t len) {
	int rc;
	struct flock fdata;
	fdata.l_type = F_WRLCK;
	fdata.l_whence = SEEK_SET;
	fdata.l_start = 0;
	fdata.l_len = 0;
	// fdata.l_sysid=0;
	// fdata.l_pid=0;
	repeat: if ((rc = fcntl(rendezvousfd, F_SETLKW, &fdata)) < 0) {
		if (errno == EINTR) {
			goto repeat;
			// 8-)
		} else {
			perror("a_writelock");
			exit(EX_OSERR);
		}
	}
	return (rc);
}

void b_writelock(void) {
	a_writelock(0, 0);
}

void c_writelock(int group, int channel) {
	a_writelock(((char *) &(grp_channel(common, group, channel)))
			- (char *) common, sizeof(CHANNEL));
}

int a_unlock(off_t start, off_t len) {
	int rc;
	struct flock fdata;
	fdata.l_type = F_UNLCK;
	fdata.l_whence = SEEK_SET;
	fdata.l_start = 0;
	fdata.l_len = 0;
	// fdata.l_sysid=0;
	// fdata.l_pid=0;
	repeat: if ((rc = fcntl(rendezvousfd, F_SETLK, &fdata)) < 0) {
		if (errno == EINTR) {
			goto repeat;
			// 8-)
		} else {
			perror("a_unlock");
			exit(EX_OSERR);
		}
	}
	return (rc);
}

void b_unlock(void) {
	a_unlock(0, 0);
}

void c_unlock(int group, int channel) {
	a_unlock(((char *) &(grp_channel(common, group, channel))) - (char *) common, sizeof(CHANNEL));
}

void *shm_malloc(char *file, int size) {
	char *data = NULL;
	key_t key;
	int shmid;

	shmfilefd = open(file, O_RDWR | O_CREAT, 0644);
	if (shmfilefd < 0) {
		fatal_err_dump("Cannot open shared memory file");
	}

	if (ftruncate(shmfilefd, size) < 0) {
		close(shmfilefd);
		fatal_err_dump("Cannot set file size of shared mem file");
	}

	data = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, shmfilefd, 0);
	if (!data || data == MAP_FAILED) {
		close(shmfilefd);
		fatal_err_dump("Cannot map shared data to shared mem file");
	}

	return (data);
}

static unsigned int hash_fold(char* s, int len) {
	unsigned int rc = 0;
	int i = 0;
	for (i = 0; i < len; i++) {
		rc = s[i] + 31 * rc;
	}
	return (rc);
}

static ssize_t writen(int fd, unsigned char *ptr, size_t nbytes) {
	int nleft;
	ssize_t nwritten;

	nleft = nbytes;

	while (nleft > 0) {
		nwritten = write(fd, ptr, nleft);
		if (nwritten <= 0) {
			return (nwritten); /* error */
		}
		nleft -= nwritten;
		ptr += nwritten;
	}

	return (nbytes - nleft);
}

int init_serversocket(char* socket_path) {
	int srv_socket, status, sockopton;
	struct sockaddr_un server;
	int len;

	srv_socket = socket(AF_UNIX, SOCK_STREAM, 0);
	if (srv_socket < 0) {
		fatal_err_dump("socket()");
	}

	/* configure balance's socket, make it reusable and define buffers' length in both directions*/
	//solves "address already in use"
	sockopton = 1;
	setsockopt(srv_socket, SOL_SOCKET, SO_REUSEADDR, (char*) &sockopton,
			sizeof(sockopton));
	//setsockopt(srv_socket, SOL_SOCKET, SO_SNDBUF, &sockbufsize, sizeof(sockbufsize));
	//setsockopt(srv_socket, SOL_SOCKET, SO_RCVBUF, &sockbufsize, sizeof(sockbufsize));

	server.sun_family = AF_UNIX;
	strcpy(server.sun_path, socket_path);
	unlink(server.sun_path);
	len = strlen(server.sun_path) + sizeof(server.sun_family);

	status = bind(srv_socket, (struct sockaddr *) &server, len);
	if (status < 0) {
		fatal_err_dump("bind()");
	}

	status = listen(srv_socket, BALANCE_QUEUE_LENGTH);
	if (status < 0) {
		fatal_err_dump("listen()");
	}
	return (srv_socket);
}

void init_worker_channel(char* chanel_path, int *maxc, char* argv_i) {
	char *p;
	/* handle unix socket path*/
	if ((p = strtok(argv_i, ":")) == NULL) {
		fatal_err_dump("worker unix socket path should be specified");
	} else {
		strcpy(chanel_path, workers_path);
		strcat(chanel_path, p);
	}
	/* handle max connection for worker*/
	if ((p = strtok(NULL, ":")) == NULL) {
		*maxc = 0;
	} else {
		*maxc = atoi(p);
	}
	if (debugflag) {
		fprintf(stderr, "worker: %s,%d\n", chanel_path, *maxc);
	}

	if (debugflag) {
		struct sockaddr_un remote;
		int s;
		if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
			perror("socket error to worker");
			exit(1);
		}

		remote.sun_family = AF_UNIX;
		strcpy(remote.sun_path, chanel_path);
		int len = strlen(remote.sun_path) + sizeof(remote.sun_family);
		if (connect(s, (struct sockaddr *) &remote, len) == -1) {
			perror("connect error to worker");
			exit(1);
		} else {
			close(s);
		}
	}
}

int balance(int* last_group_index, struct in_addr* cli_addr,
		enum boolean_t from_master) {
	/* index over groups */
	int start_group_index = *last_group_index;
	int current_group_index = start_group_index;
	/* index over channels in group */
	int current_channel = -1;
	/* error code for transaction*/
	int error_code = 0;

	int cicle_counter = 0;

	b_writelock();
	//for over groups
	for (;;) {
		//start finding place in current group from next channel to current one or from 0 if it's first time
		if (current_group_index == -1) {
			//supposing there is always 0 group present with one channel
			current_group_index = 0;
			current_channel = 0;
		} else {
			//try next channel in same group
			current_channel = (grp_current(common, current_group_index) + 1)
					% grp_nchannels(common, current_group_index);
		}

		//for over channels in group
		for (;;) {
			if (grp_type(common, current_group_index) == GROUP_RR) {
				/* if there is place in current channel take it*/
				if (chn_status(common, current_group_index, current_channel)
						== CHANNEL_ENABLED
						&& (chn_maxc(common, current_group_index, current_channel)
								== 0
								|| (chn_c(common, current_group_index, current_channel)
										< chn_maxc(common, current_group_index, current_channel)))) {
					break;
				}
				//move to next channel if current is full
				else {
					current_channel++;
					//mod # of channels in group
					if (current_channel
							>= grp_nchannels(common, current_group_index)) {
						current_channel = 0;
					}
					//we made full cicle
					if (current_channel
							== grp_current(common, current_group_index)) {
						current_channel = -1; // no channel available in this group
						break;
					}
				}

			} else if (grp_type(common,current_group_index) == GROUP_HASH) {
				unsigned int uindex = hash_fold((char *) cli_addr,
						sizeof(*cli_addr));
				current_channel = uindex
						% grp_nchannels(common, current_group_index);
				//if there is place in current channel
				if (chn_status(common, current_group_index, current_channel)
						== CHANNEL_ENABLED
						&& (chn_maxc(common, current_group_index, current_channel)
								== 0
								|| (chn_c(common, current_group_index, current_channel)
										< chn_maxc(common, current_group_index, current_channel)))) {
					break;
				}//choose another group
				else {
					//hash failover is done by RR technique from current_channel in current group
					if (hashfailover == 1) {
						int startindex = current_channel;
						for (;;) {
							current_channel++;
							//mod
							if (current_channel
									>= grp_nchannels(common, current_group_index)) {
								current_channel = 0;
							}
							//made whole cicle in current group
							if (current_channel == startindex) {
								current_channel = -1;
								break;
							}
							//check if channel isn't full
							if (chn_status(common, current_group_index, current_channel)
									== CHANNEL_ENABLED
									&& (chn_maxc(common, current_group_index, current_channel)
											== 0
											|| (chn_c(common, current_group_index, current_channel)
													< chn_maxc(common, current_group_index, current_channel)))) {
								break;
							}
						}
					} else {
						current_channel = -1;
					}
					break;
				}
			}//wrong group type
			else {
				error_code = 1;
				break;
			}
		}//for over channels

		if (current_channel < 0) {
			// try next group starting from 0 channel
			current_group_index = (current_group_index + 1)
					% cmn_ngroups(common);
			current_channel = 0;

			if (current_group_index == start_group_index) {
				if (cicle_counter == max_retries) {
					if (debugflag) {
						fprintf(
								stderr,
								"start_group, current_group, current_channel %d %d %d",
								start_group_index, current_group_index,
								current_channel);
					}
					current_group_index = -1;
					current_channel = -1;
					break;
				} else {
					cicle_counter++;
					continue;
				}
			}
		}/* we found legal channel in current group */
		else {
			break;
		}
	}//for over groups

	/* all groups and all channels are full*/
	if (current_channel < 0 || current_group_index < 0) {
		cmn_fc(common)++;
		error_code = 2;
	}
	/* we found place in current group */
	else {
		//remember group where we find empty channel so next time start from it
		if (from_master == TRUE) {
			*last_group_index = current_group_index;
		}

		//remember current channel in group
		grp_current(common, current_group_index) = current_channel;

		//reserve channel for client(still unsuccessful connection is possible)
		chn_c(common, current_group_index, current_channel)++;
		chn_tc(common, current_group_index, current_channel)++;

	}
	b_unlock();

	//tanger_commit();
	switch (error_code) {
	//tanger_thread_shutdown();
	//tanger_shutdown();
	case 1:
		fatal_err_dump("PANIC: invalid group type \n");
		break;
	case 2:
		if (debugflag) {
			fprintf(stderr, "all groups and all channels are full \n");
		}
		break;
	}

	return current_channel;
}//end of balance

void update_bytes_on_channel(int groupindex, int channelindex, ssize_t rc,
		enum boolean_t is_forward) {
	//tanger_begin();
	c_writelock(groupindex, channelindex);
	if (is_forward == TRUE) {
		chn_bsent(common, groupindex, channelindex) += rc;
	} else {
		chn_breceived(common, groupindex, channelindex) += rc;
	}
	c_unlock(groupindex, channelindex);
	//tanger_commit();
}

int relay(int fromfd, int tofd, int groupindex, int channelindex,
		enum boolean_t is_forward) {
	ssize_t rc;
	char* print_symbol = (is_forward) ? "->" : "<-";
	unsigned char buffer[MAX_MESSAGE_SIZE];
	bzero(buffer, MAX_MESSAGE_SIZE);
	rc = read(fromfd, buffer, MAX_MESSAGE_SIZE);

	//we ended forwarding data
	if (rc <= 0) {
		return (-1);
	} else {
		if (packetdump) {
			fprintf(stderr, "%s %s, %d\n", print_symbol, buffer, rc);
		}
		if (writen(tofd, buffer, rc) != rc) {
			return (-1);
		}
		update_bytes_on_channel(groupindex, channelindex, rc, is_forward);
	}
	return (0);
}

/*
 * the connection is really established, let's transfer the data
 *  as efficient as possible :-)
 */
void relay_stream(int clientfd, int worker_socketfd, int group_index,
		int channel_index) {
	//client_worker_fds will be modified to reflect which of the file descriptors you selected which is ready for reading
	fd_set client_worker_fds;
	fd_set read_fds; // temp file descriptor list for select()
	int fdset_width;
	int sr;
	int optone = 1;

	//The parameter fdset_width should be set to the values of the highest file descriptor plus one
	fdset_width = ((clientfd > worker_socketfd) ? clientfd : worker_socketfd)
			+ 1;

	/* before relaying configure woker's socket*/
	//setsockopt(worker_socketfd, IPPROTO_TCP, TCP_NODELAY, (char *) &optone, (socklen_t) sizeof(optone));
	//setsockopt(worker_socketfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &optone, (socklen_t) sizeof(optone));
	//configure worker socket
	//setsockopt(worker_sockfd, SOL_SOCKET, SO_SNDBUF, &sockbufsize, sizeof(sockbufsize));
	//setsockopt(worker_sockfd, SOL_SOCKET, SO_RCVBUF, &sockbufsize, sizeof(sockbufsize));

	/* configure client's socket */
	//setsockopt(clientfd, IPPROTO_TCP, TCP_NODELAY, (char *) &optone, (socklen_t) sizeof(optone));
	//setsockopt(clientfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &optone, (socklen_t) sizeof(optone));

	//Clear all entries from the set
	FD_ZERO(&client_worker_fds);
	FD_ZERO(&read_fds);
	//Add clientfd to the set
	FD_SET(clientfd, &client_worker_fds);
	//Add workerfd to the set
	FD_SET(worker_socketfd, &client_worker_fds);

	//relay for
	for (;;) {
		/*
		 * just in case this system modifies the timeout values or previous select modifies read_fds,
		 * refresh the values from a saved copy of them.
		 */
		read_fds = client_worker_fds;
		sel_tmout = save_tmout;

		//select on open socket
		for (;;) {

			if (sel_tmout.tv_sec || sel_tmout.tv_usec) {
				sr = select(fdset_width, &read_fds, NULL, NULL, &sel_tmout);
			} else {
				sr = select(fdset_width, &read_fds, NULL, NULL, NULL);
			}
			if ((save_tmout.tv_sec || save_tmout.tv_usec) && !sr) {
				fprintf(stderr, "timed out after %d seconds\n",
						(int) save_tmout.tv_sec);
				fflush(stderr);
				exit(EX_UNAVAILABLE);
			}
			//EINTR interuption by some signal
			if (sr < 0 && errno != EINTR) {
				fatal_err_dump("select error");
			}
			/* something present over sockets*/
			if (sr > 0) {
				break;
			}
		}//end of for select open client/worker sockets

		//if client sent data forward it to worker
		if (FD_ISSET(clientfd, &read_fds)) {
			if (relay(clientfd, worker_socketfd, group_index, channel_index,
					TRUE) < 0) {
				break;
			}
		}//otherwise backward data from worker to client
		if (FD_ISSET(worker_socketfd, &read_fds)) {
			if (relay(worker_socketfd, clientfd, group_index, channel_index,
					FALSE) < 0) {
				break;
			}
		}
	}//end of relay for
}

/*
 * tries to connect to selected channel in group
 * after successful connection delegates stream relay to stream_relay function
 */
void* handle_client(int clientfd, int group_index, int channel_index, struct in_addr client_address, enum boolean_t is_one_worker) {
	struct sockaddr_un worker_addr;
	worker_addr.sun_family = AF_UNIX;

	if (debugflag) {
		fprintf(stderr, "working with group:channel %d:%d", group_index, channel_index);
	}
	int worker_sockfd = 0;
	//connection for to worker server
	for (;;) {
		if (debugflag) {
			fprintf(stderr, "trying to connect to worker in group %d channel %d \n", group_index, channel_index);
		}
		if ((worker_sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
			fatal_err_dump("can't open stream socket");
		}

		//no need to synch, since chn_path is read-only after init
		strcpy(worker_addr.sun_path, chn_path(common, group_index, channel_index));

		if (connect(worker_sockfd, (struct sockaddr *) &worker_addr, sizeof(worker_addr)) < 0) {
			close(worker_sockfd);
			if (errno == EINTR) {
				fprintf(stderr, "timeout group %d channel %d\n", group_index, channel_index);
			} else {
				fprintf(stderr, "connection refused group %d channel %d\n",
						group_index, channel_index);
			}

			/* reservation broken, clean the channel */
			//tanger_begin();
			c_writelock(group_index, channel_index);
			chn_c(common, group_index, channel_index)--;
			chn_tc(common, group_index, channel_index)--;
			c_unlock(group_index, channel_index);
			//tanger_commit();

			//find another group and channel
			fprintf(stderr, "rebalancing...");
			if ((channel_index = balance(&group_index, &client_address, FALSE)) > 0) {
				continue;
			} else {
				close(worker_sockfd);
				close(clientfd);
				return;
			}
		}
		// we made successful connection to worker server
		else {
			if (debugflag) {
				fprintf(stderr,	"successful connect on reserved group %d, channel %d \n", group_index, channel_index);
			}
			//reservation is correct, connection succeeded, proceed to relay streams
			relay_stream(clientfd, worker_sockfd, group_index, channel_index);
			break;
		}
	}//end of connection for to worker server


	if (debugflag) {
		fprintf(stderr, "releasing group:channel %d:%d \n\n ", group_index, channel_index);
	}
	/* after all work done free the channel*/
	c_writelock(group_index, channel_index);
	chn_c(common, group_index, channel_index)--;
	c_unlock(group_index, channel_index);

	close(worker_sockfd);
	close(clientfd);
	return;
}

static void usage(void) {
	fprintf(stderr, "  usage:\n");
	fprintf(stderr,	"  balance [-b balance-unix-socket-path] [-t sec] [-T sec] [-dfpH] \\\n");
	fprintf(stderr,	"          [path[:maxc1]] [!%%] [ ... hN[:pN[:maxcN]]]]\n");
	fprintf(stderr, "\n");
	fprintf(stderr, "  -c #cpus:cpu to bind to\n");
	fprintf(stderr, "  -s #path to workers directory\n");
	fprintf(stderr, "  -b path   bind to specific socket-path to listen\n");
	fprintf(stderr, "  -d        debugging on\n");
	fprintf(stderr, "  -f        stay in foregound\n");
	fprintf(stderr, "  -H        failover even if Hash Type is used\n");
	fprintf(stderr, "  -p        packet dump \n");
	fprintf(stderr,	"  -T sec    timeout (seconds) for select (0 => never) (default=%d)\n", DEFAULT_SELECT_TIMEOUT);
	fprintf(stderr,	"   !        separates channel groups (declaring previous to be Round Robin)\n");
	fprintf(stderr,	"   %       as !, but declaring previous group to be a Hash Type\n");
	fprintf(stderr, "\n");
	fprintf(stderr, "example:\n");
	fprintf(stderr, "balance -b ./sockets/balance -t 10 -T 10 -d -f -p -H ./sockets/w1:10 ./sockets/w2 ! ./sockets/w3:100 %% \n");
	fprintf(stderr, "\n");

	exit(EX_USAGE);
}
// goto background:

void background(void) {
	int childpid;
	if ((childpid = fork()) < 0) {
		fprintf(stderr, "cannot fork\n");
		exit(EX_OSERR);
	} else {
		if (childpid > 0) {
			exit(EX_OK); /* parent */
		}
	}
	setpgrp();
	chdir("/");
	close(0);
	close(1);
	close(2);
}

/* done in main thread, no need to synch*/
COMMON *makecommon(int argc, char **argv) {
	int i;
	int group;
	int channel;
	COMMON *mycommon;
	struct stat buffer;
	int fd;

	if (stat(rendezvousfile, &buffer) == -1) {
		// File not existing yet ...
		if ((fd = open(rendezvousfile, O_CREAT | O_RDWR, 0666)) == -1) {
			fprintf(stderr, "cannot create rendezvous file %s\n",
					rendezvousfile);
			exit(EX_OSERR);
		} else {
			if (debugflag)
				fprintf(stderr, "file %s created\n", rendezvousfile);
			close(fd);
		}
	} else {
		if (debugflag)
			fprintf(stderr, "file %s already exists\n", rendezvousfile);
	}

	if ((rendezvousfd = open(rendezvousfile, O_RDWR, 0)) < 0) {
		perror("open");
		fprintf(stderr, "check rendezvousfile permissions [%s]\n", rendezvousfile);
		exit(EX_NOINPUT);
	}

	if ((mycommon = (COMMON *) shm_malloc(rendezvousfile, sizeof(COMMON))) == NULL) {
		fprintf(stderr, "cannot alloc COMMON struct\n");
		exit(EX_OSERR);
	}

	mycommon->fc = 0;
	mycommon->pid = getpid();

	fprintf(stderr,  "pid: %d", mycommon->pid);

	for (group = 0; group < MAX_GROUPS; group++) {
		grp_nchannels(mycommon, group) = 0;
		grp_current(mycommon, group) = 0;
		grp_type(mycommon, group) = GROUP_RR; // Default: RR
	}

	group = 0;
	channel = 0;

	for (i = 0; i < argc; i++) {
		/*end of round robin group*/
		if (strcmp(argv[i], "!") == 0) {
			if (channel <= 0) {
				fatal_err_dump("no channels in group");
			}
			grp_type(mycommon, group) = GROUP_RR;
			grp_nchannels(mycommon, group) = channel;

			/* proceed to next group*/
			group++;
			channel = 0;
			if (group >= MAX_GROUPS) {
				fatal_err_dump("too many groups");
			}
		}/* end of hash group*/
		else if (strcmp(argv[i], "%") == 0) {
			if (channel <= 0) {
				fatal_err_dump("no channels in group");
			}
			grp_type(mycommon, group) = GROUP_HASH;
			grp_nchannels(mycommon, group) = channel;

			/* proceed to next group*/
			group++;
			channel = 0;
			if (group >= MAX_GROUPS) {
				fatal_err_dump("too many groups");
			}
		} else {
			chn_status(mycommon, group, channel) = CHANNEL_ENABLED;
			chn_c(mycommon, group, channel) = 0; // connections...
			chn_tc(mycommon, group, channel) = 0; // total connections...
			init_worker_channel(chn_path(mycommon, group, channel),
					&chn_maxc(mycommon, group, channel), argv[i]);
			chn_bsent(mycommon, group, channel) = 0;
			chn_breceived(mycommon, group, channel) = 0;

			channel++;
			if (channel >= MAX_CHANNELS) {
				fatal_err_dump("too many channels in one group");
			}
		}
	}
	/* in a case user didn't give group type */
	if (channel > 0) {
		grp_type(mycommon, group) = GROUP_RR;
		grp_nchannels(mycommon, group) = channel;
		group++;
	}
	cmn_ngroups(mycommon) = group;

	if (debugflag) {
		fprintf(stderr, "number of active groups: %d\n", cmn_ngroups(mycommon));
		for (group = 0; group < cmn_ngroups(mycommon); group++) {
			fprintf(stderr, "number of channels %d \n",
					grp_nchannels(mycommon, group));
			for (i = 0; i < grp_nchannels(mycommon, group); i++) {
				fprintf(stderr, "%d %d %s:%d\n", group, i,
						chn_path(mycommon, group, i),
						chn_maxc(mycommon, group, i));
				fflush(stderr);
			}
		}
	}

	return (mycommon);
}

void handle_affinity(char* cpu_arg){
		int pid = getpid();
		if (cpu_arg != NULL) {
		fprintf(stderr, "cpu arg %s\n", cpu_arg);
		cpu_set_t* cpu_set;
		char* p = NULL;
		int num_of_cpus = 0;
		if ((p = strtok(cpu_arg, ":")) == NULL) {
			perror("split error");
			return;
		} else {
			num_of_cpus = atoi(p);
			fprintf(stderr, "num of cpus %d\n", num_of_cpus);
			cpu_set = CPU_ALLOC(num_of_cpus);
		}
		size_t size = CPU_ALLOC_SIZE(num_of_cpus);

		CPU_ZERO_S(size, cpu_set);
		while ((p = strtok(NULL, ",")) != NULL) {
			int cpu_num = atoi(p);
			fprintf(stderr, "cpu_num to run on %d\n", cpu_num);
			CPU_SET_S(cpu_num, size, cpu_set);
		}

		fflush(stderr);

		if (sched_setaffinity(pid, size, cpu_set) < 0) {
			perror("sched_setaffinity");
		}

		if (sched_getaffinity(pid, size, cpu_set) < 0) {
			perror("sched_getaffinity");
			return -1;
		}
		fprintf(stderr, "pid %d's affinity mask: %d\n", pid, *cpu_set);
		CPU_FREE(cpu_set);
	}
}

int main(int argc, char *argv[]) {
	/* used to parse options*/
	int c;
	while ((c = getopt(argc, argv, "c:b:T:s:dfpHo")) != EOF) {
				switch (c) {
				case 'c':
					handle_affinity(optarg);
					break;
				case 'b':
					balance_socket_path = optarg;
					break;
				case 'T':
					sel_tmout.tv_sec = atoi(optarg);
					sel_tmout.tv_usec = 0;
					if (sel_tmout.tv_sec < 1)
						usage();
					save_tmout = sel_tmout;
					break;
				case 's':
					strcpy(workers_path,optarg);
					break;
				case 'f':
					foreground = 1;
					break;
				case 'd':
					debugflag = 1;
					break;
				case 'p':
					packetdump = 1;
					break;
				case 'H':
					hashfailover = 1;
					break;
				case 'o':
					is_one_worker = TRUE;
					break;
				case '?':
				default:
					usage();
				}
			}

	argc -= optind;
	argv += optind;

	if (debugflag) {
		fprintf(
				stderr,
				"balance_socket_path %s : sel_tmout.tv_sec %ld :foreground %d : debugflag %d : packetdump %d : hashfailover %d ",
				balance_socket_path, sel_tmout.tv_sec, foreground, debugflag,
				packetdump, hashfailover);
		fprintf(stderr, "after parsing options: \n");
		fprintf(stderr, "argv[0]=%s\n", argv[0]);
		fprintf(stderr, "argc : %d\n", argc);
	}

	if (!foreground) {
		background();
	}

	/*master doesn't need to wait for children, very! important*/
	signal(SIGCHLD, SIG_IGN);
	if (signal(SIGUSR1, sig_handler) == SIG_ERR) {
		perror("signal");
		exit(1);
	}

	// Open a server's UNIX socket
	int sockfd = init_serversocket(balance_socket_path);

	/* create shared groups'/channels' data*/
	common = makecommon(argc, argv);

	/* start balancing from 0 group, multi-reader var */
	int last_group_index = -1;

	for (;;) {
		char client_id_str[SIZE_OF_ID];
		bzero(client_id_str, SIZE_OF_ID);

		struct sockaddr_un cli_addr;
		/* length of client socket structure*/
		int clilen = sizeof(cli_addr);
		int new_client_socket = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
		client_connects++;
		if (new_client_socket < 0) {
			if (debugflag) {
				fprintf(stderr, "accept error %d\n", errno);
				fflush(stderr);
				exit(1);
			}
			continue;
		}
		/* first INET_ADDRSTRLEN bytes of packet is id(or ip) of client */
		int n = recv(new_client_socket, (&client_id_str), SIZE_OF_ID, 0);
		if (n <= 0) {
                        perror("recv error, id isn't sent");
                        close(new_client_socket);
                        continue;

		}

		/* client address for local unix-socket*/
		struct in_addr client_id;
		if (inet_aton(client_id_str, &client_id) < 0) {
			fatal_err_dump("inet_ntoa error");
		}
		if (client_id.s_addr < 0) {
			fatal_err_dump("illegal client id");
		}
		if (debugflag) {
			fprintf(stderr, "new connection from %s read n: %d bytes \n", client_id_str, n);
			u_int32_t ip = client_id.s_addr;
			fprintf(stderr, "%d.%d.%d.%d\n", (ip >> 24) & 0xFF, (ip >> 16) & 0xFF, (ip >> 8) & 0xFF, ip & 0xFF);
		}

		int allocated_channel = -1;
		/* balance from group where place was last time found*/
		if ((allocated_channel = balance(&last_group_index, &client_id, TRUE)) >= 0) {
			if (debugflag) {
				fprintf(stderr, "allocated %d group and %d channel\n", last_group_index, allocated_channel);
			}

			int childpid = -1;
			if ((childpid = fork()) < 0) {
				// the connection is rejected if fork() returns error,
				// but main process stays alive !
				if (debugflag) {
					if(childpid == EAGAIN){
						perror("fork error: # of concurrent processes reached\n");
					}else if(childpid == ENOMEM){
						perror("fork error: insufficient storage space\n");
					}else{
						perror("fork error: other problem\n");
					}
				}
			} else if (childpid == 0) { // child process
				close(sockfd); // close original socket
				// process the request:
				handle_client(new_client_socket, last_group_index, allocated_channel, client_id, is_one_worker);
				exit(EX_OK);
			} else {
				close(new_client_socket); // parent process
			}
		} else {
			close(new_client_socket);
		}
	}//end of accept for
}
